package org.acme.cdc.factory;

import com.github.shyiko.mysql.binlog.BinaryLogClient;
import com.github.shyiko.mysql.binlog.event.*;
import org.acme.cdc.annotation.CDCListener;
import org.acme.cdc.constant.CDCConstant;
import org.acme.cdc.wrap.BinlogAddObj;
import org.acme.cdc.wrap.BinlogDeleteObj;
import org.acme.cdc.wrap.BinlogObj;
import org.acme.cdc.wrap.BinlogUpdateObj;
import org.acme.cdc.listener.CDCMessageListener;
import org.acme.cdc.service.SysColService;
import org.acme.server.service.SysServerService;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.beans.BeansException;
import org.springframework.beans.factory.DisposableBean;
import org.springframework.beans.factory.InitializingBean;
import org.springframework.beans.factory.annotation.Value;
import org.springframework.beans.factory.config.BeanPostProcessor;
import org.springframework.stereotype.Service;

import javax.annotation.Resource;
import java.io.Serializable;
import java.util.BitSet;
import java.util.HashMap;
import java.util.List;
import java.util.Map;
import java.util.concurrent.ConcurrentHashMap;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;

/*
 * @className: CDCServerFactory
 * @author: chentao
 * @date: 2025/3/7 下午2:34
 * @Version: 1.0
 * @description:
 */
@Service
public class CDCServerFactory implements BeanPostProcessor, InitializingBean, DisposableBean, BinaryLogClient.EventListener, BinaryLogClient.LifecycleListener {

    Logger logger = LoggerFactory.getLogger(getClass());

    private Map<Long, String> tableMap = new ConcurrentHashMap<>();

    private Map<String, CDCMessageListener> listenerMap = new ConcurrentHashMap<>();

    @Value("${spring.datasource.host}")
    private String host;

    @Value("${spring.datasource.port}")
    private int port;

    @Value("${spring.datasource.db}")
    private String db;

    @Value("${spring.datasource.username}")
    private String username;

    @Value("${spring.datasource.password}")
    private String password;

    @Resource
    private SysServerService sysServerService;

    private BinaryLogClient binaryLogClient;

    private ExecutorService executorService;

    @Resource
    private SysColService sysColService;

    @Override
    public Object postProcessAfterInitialization(Object bean, String beanName) throws BeansException {
        if (bean.getClass().isAnnotationPresent(CDCListener.class) && bean instanceof CDCMessageListener) {
            CDCListener dbListener = bean.getClass().getAnnotation(CDCListener.class);
            this.listenerMap.put(dbListener.tableName(), (CDCMessageListener) bean);
        }
        return BeanPostProcessor.super.postProcessAfterInitialization(bean, beanName);
    }

    /****
     * 初始化并启动
     * @throws Exception
     */
    @Override
    public void afterPropertiesSet() throws Exception {
        long serverId = sysServerService.getServerId();
        this.executorService = Executors.newCachedThreadPool();
        this.binaryLogClient = new BinaryLogClient(this.host, this.port, this.db, this.username, this.password);
        this.binaryLogClient.setServerId(serverId);
        this.binaryLogClient.registerEventListener(this);
        this.binaryLogClient.registerLifecycleListener(this);
        this.executorService.execute(new ServerStartThread());
    }

    /****
     * 删除并停止链接
     * @throws Exception
     */
    @Override
    public void destroy() throws Exception {
        this.binaryLogClient.disconnect();
        if (!this.executorService.isShutdown()) {
            this.executorService.shutdownNow();
        }
    }

    /***
     * 数据库表变化的消息监听
     * @param event
     */
    @Override
    public void onEvent(Event event) {
        if (event.getData() instanceof TableMapEventData) {
            TableMapEventData eventData = (TableMapEventData) event.getData();
            long tableId = eventData.getTableId();
            String tableName = eventData.getTable();
            this.tableMap.putIfAbsent(tableId, tableName);
        } else {
            BinlogObj binlogObj = this.wrapBinlogObj(event);
            if (binlogObj != null && this.listenerMap.containsKey(binlogObj.getTableName())) {
                this.executorService.execute(new CDCHandlerThread(binlogObj));
            }
        }
    }

    /***
     * 封裝日志結果
     * @return
     */
    private BinlogObj wrapBinlogObj(Event event) {
        BinlogObj binlogObj = null;
        //数据的更新
        if (event.getData() instanceof UpdateRowsEventData) {
            UpdateRowsEventData updateRowsEventData = event.getData();
            String tableName = this.tableMap.get(updateRowsEventData.getTableId());
            binlogObj = new BinlogUpdateObj(tableName, CDCConstant.UPDATE, updateRowsEventData);
            if (this.listenerMap.containsKey(tableName)) {
                Map<Long, String> cols = sysColService.getColsMap(db, tableName);
                for (Map.Entry<Serializable[], Serializable[]> row : updateRowsEventData.getRows()) {
                    Map<String, Map<Serializable, Serializable>> colMap = new HashMap<>();
                    Serializable[] key = row.getKey();
                    Serializable[] value = row.getValue();
                    for (int i = 0; i < key.length; i++) {
                        Map<Serializable, Serializable> valMap = new HashMap<>();
                        String colName = cols.get((long) i);
                        valMap.put(key[i], value[i]);
                        colMap.put(colName, valMap);
                    }
                    ((BinlogUpdateObj) binlogObj).getUpdateRowList().add(colMap);
                }
            }
        }
        //数据的新增
        if (event.getData() instanceof WriteRowsEventData) {
            WriteRowsEventData writeRowsEventData = event.getData();
            String tableName = this.tableMap.get(writeRowsEventData.getTableId());
            binlogObj = new BinlogAddObj(tableName, CDCConstant.ADD, event.getData());
            if (this.listenerMap.containsKey(tableName)) {
                Map<Long, String> cols = sysColService.getColsMap(db, tableName);
                BitSet includedColumns = writeRowsEventData.getIncludedColumns();
                for (Serializable[] row : writeRowsEventData.getRows()) {
                    Map<String, Serializable> map = new HashMap<>();
                    for (int i = 0; i < includedColumns.length(); i++) {
                        int colIndex = includedColumns.nextSetBit(i);
                        map.put(cols.get((long) colIndex), row[colIndex]);
                    }
                    ((BinlogAddObj) binlogObj).getRowsList().add(map);
                }
            }
        }
        //数据的删除
        if (event.getData() instanceof DeleteRowsEventData) {
            DeleteRowsEventData deleteRowsEventData = event.getData();
            String tableName = this.tableMap.get(deleteRowsEventData.getTableId());
            binlogObj = new BinlogDeleteObj(tableName, CDCConstant.DELETE, event.getData());
            if (this.listenerMap.containsKey(tableName)) {
                Map<Long, String> cols = sysColService.getColsMap(db, tableName);
                BitSet includedColumns = deleteRowsEventData.getIncludedColumns();
                List<Serializable[]> rows = deleteRowsEventData.getRows();
                for (Serializable[] row : rows) {
                    Map<String, Serializable> map = new HashMap<>();
                    for (int i = 0; i < includedColumns.length(); i++) {
                        int colIndex = includedColumns.nextSetBit(i);
                        map.put(cols.get((long) colIndex), row[colIndex]);
                    }
                    ((BinlogDeleteObj) binlogObj).getRowsList().add(map);
                }
            }
        }
        return binlogObj;
    }

    /***
     * 有客户端连接上消息监听
     * @param client the client that logged in
     */
    @Override
    public void onConnect(BinaryLogClient client) {
        logger.warn("节点: {}", client.getServerId() + " 连接成功");
    }

    /***
     * 客户端连接失败消息监听
     * @param client the client that triggered the communication failure
     * @param ex The exception that triggered the communication failutre
     */
    @Override
    public void onCommunicationFailure(BinaryLogClient client, Exception ex) {
        logger.warn("节点连接数据失败: {}", ex.getMessage());
    }

    /****
     * 序列化失败的监听
     * @param client the client that failed event deserialization
     * @param ex The exception that triggered the failutre
     */
    @Override
    public void onEventDeserializationFailure(BinaryLogClient client, Exception ex) {
        //序列化失败
    }

    /***
     * 连接关闭的消息监听
     * @param client the client that disconnected
     */
    @Override
    public void onDisconnect(BinaryLogClient client) {
        logger.warn("节点: {}", client.getServerId() + " 断开连接");
    }

    /****
     * 服务启动线程
     */
    class ServerStartThread implements Runnable {
        public void run() {
            try {
                binaryLogClient.connect();
            } catch (Exception e) {
                logger.warn("binaryLogClient客户端启动失败: " + e.getMessage());
            }
        }
    }

    /***
     * 消息监听处理类
     */
    class CDCHandlerThread implements Runnable {

        private BinlogObj binlogObj;

        public CDCHandlerThread(BinlogObj binlogObj) {
            this.binlogObj = binlogObj;
        }

        @Override
        public void run() {
            if (binlogObj != null) {
                CDCMessageListener cdcMessageListener = listenerMap.get(binlogObj.getTableName());
                cdcMessageListener.onMessage(binlogObj);
            }
        }
    }

}
